package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.OrderInfoWide1;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class MyAsyncFunction extends RichAsyncFunction<OrderInfoWide1, OrderInfoWide1> {

    private Connection conn1;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn1 = DriverManager.getConnection("jdbc:phoenix:hadoop102:2181");
    }

    @Override
    public void close() throws Exception {
        if (conn1 != null)
            conn1.close();

    }

    @Override
    public void asyncInvoke(OrderInfoWide1 orderInfoWide, ResultFuture<OrderInfoWide1> resultFuture) throws Exception {
        // 一定先关联sku
        String sql1 = "select line from dim_product_sku_yk8 where id=?";

        try {
            PreparedStatement ps1 = conn1.prepareStatement(sql1);
            ps1.setObject(1, orderInfoWide.getProduct_sku_id());
            ResultSet rs1 = ps1.executeQuery();
            if (rs1.next()) {
                String line1 = rs1.getString("line");
                if (line1 != null) {
                    // 转成JSON
                    JSONObject ob1 = JSON.parseObject(line1);

                    Long product_spu_id = ob1.getLong("product_spu_id");

                    orderInfoWide.setProduct_spu_id(product_spu_id);
                }
            }



            // 关联spu
            String sql2 = "select line from dim_product_spu_yk8 where id= ?";
            PreparedStatement ps2 = conn1.prepareStatement(sql2);
            ps2.setObject(1, orderInfoWide.getProduct_spu_id());
            ResultSet rs2 = ps2.executeQuery();
            if (rs2.next()) {
                String line2 = rs2.getString("line");
                if (line2 != null) {
                    // 转成JSON
                    JSONObject ob2 = JSON.parseObject(line2);
                    String name = ob2.getString("name");
                    orderInfoWide.setProduct_spu_name(name);
                }
            }

            // 关联店铺
            String sql3 = "select line from dim_shop_yk8 where id=?";
            PreparedStatement ps3 = conn1.prepareStatement(sql3);
            ps3.setObject(1, orderInfoWide.getShop_id());
            ResultSet rs3= ps3.executeQuery();
            if (rs3.next()) {
                String line3 = rs3.getString("line");
                if (line3 != null) {
                    // 转成JSON
                    JSONObject ob3 = JSON.parseObject(line3);
                    Integer regionId = ob3.getInteger("region_id");
                    orderInfoWide.setRegion_id(regionId);
                }
            }

            // 关联地区
            String sql4 = "select line from dim_region_yk8 where id=?";
            PreparedStatement ps4 = conn1.prepareStatement(sql4);
            ps4.setInt(1, orderInfoWide.getRegion_id());
            ResultSet rs4= ps4.executeQuery();
            if (rs4.next()) {
                String line4 = rs4.getString("line");
                if (line4 != null) {
                    // 转成JSON
                    JSONObject ob4 = JSON.parseObject(line4);
                    String name = ob4.getString("name");
                    orderInfoWide.setRegion_name(name);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

        // 关联店铺
        resultFuture.complete(Collections.singleton(orderInfoWide));

    }

}
